package edu.gmu.router;

import edu.gmu.middleware.DispatchEntry;
import edu.gmu.middleware.ByteBufferPool;
import edu.gmu.middleware.Header;
import edu.gmu.middleware.Marshaler;
import edu.gmu.middleware.MarshalerImpl;
import edu.gmu.middleware.Message;
import edu.gmu.middleware.MessageBinding;
import edu.gmu.middleware.MessageRegistry;
import edu.gmu.middleware.Sink;
import edu.gmu.middleware.SinkDataEvent;
import edu.gmu.middleware.SinkDisconnectedEvent;
import edu.gmu.middleware.SinkListener;
import edu.gmu.middleware.Unmarshaler;
import edu.gmu.middleware.UnmarshalerImpl;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * The RouterDispatcher...
 * 
 * @author  James H. Pope
 * @version $Revision:$ $Date:$
 */
public class RouterDispatcher implements SinkListener
{
    private volatile boolean stopped = false;
    private List dispatchListeners   = new ArrayList();
    
    private Lock lock                = new ReentrantLock(true); // make fair
    private Condition messageReady   = lock.newCondition();
    private List dispatchQueue       = new ArrayList();
    
    private Marshaler marshaler      = new MarshalerImpl();
    private Unmarshaler unmarshaler  = new UnmarshalerImpl();
    
    private ByteBufferPool inPool    = null;
    private ByteBuffer out           = null;
    
    private MessageRegistry registry = new MessageRegistry();
    
    /**
     * Creates a new RouterDispatcher specified values.
     * @param inPoolSize
     * @param inBufferSize
     */
    public RouterDispatcher( int inPoolSize, int inBufferSize)
    {
        /*
         *  NOTE: BIG ALLOCATIONS HERE
         */
        this.inPool = new ByteBufferPool(inPoolSize,inBufferSize);
        // Write has own buffer so cannopt fail due to lack of buffer
        this.out    = ByteBuffer.allocate( inBufferSize );
    }
    
    /**
     * Registers binding between given id and class
     * @param id
     * @param class to bind to id
     * @returns the newly registered MessageRegistry
     * @throws IllegalArgumentException if id is already bound
     */
    public MessageBinding register( int id, Class boundClass )
    {
        return this.registry.register(id, boundClass);
    }

    /**
     * Marshals message and writes to sink - both are carried out on
     * the caller's thread.
     * @param header
     * @param message
     * @param sink
     * @throws IOException if the unmashaling/write process was not successful
     */
    public void send( Header header, Message message, Sink sink ) throws IOException
    {
        //Header header = new Header();
        this.marshaler.marshal(header, message, out);
        sink.write(out);
    }
    
    
    /**
     * Called from source object when message is received.
     * @param event
     * @throws IOException if no free buffer to copy data into
     */
    public void dataReceived( SinkDataEvent event )
    {
        //----------------------------------------------------------------//
        // Note that we copy on caller's thread and unmarshal on dispatchThread
        //----------------------------------------------------------------//
        /*
         * Make copy and return as quickly as possible.
         */
        // NEED TO REASON - WHY FAILES WHEN THIS IS INSIDE LOCK????
        ByteBuffer in = event.getIn();
        
        //System.out.println("Receiving");
        //print( in, 12 );

        ByteBuffer inBuffer = this.inPool.checkout();
        if( inBuffer == null )
        {
            String s = "No free buffer in inPool to unmarshal message - dropping.";
            System.out.println( s );
            //IOException e = new IOException(s);
            //throw e;
            //this.fireExceptionReceived(e);
            return;
        }

        /*
         * We have a free buffer - go ahead and copy in to the inPool
         * buffer - we will unmarshal on a different thread
         * after which the inPool will be returned to the pool
         */
        inBuffer.clear();
        inBuffer.put(in); // Note: System.arrayCopy may be more efficient???
        inBuffer.flip();
        //System.out.println("Dispatching");
        //print( inBuffer, 12 );

        this.lock.lock();
        try
        {
            this.dispatchQueue.add( new DispatchEntry(  event.getHeader(),
                                                        inBuffer,
                                                        event.getSink()) );
            this.messageReady.signal();
        }
        finally
        {
            this.lock.unlock();
        }
        //----------------------------------------------------------------//
    }
    
    /**
     * Called from source object when the sink is disconnected.
     * @param event
     */
    public void sinkDisconnected( SinkDisconnectedEvent event )
    {
        // Regardless of disconnect type - go ahead and shutdown
        // Another option is if exception, try to reconnect
        //--------------------------------------------------------------------//
        // Called on Sink readerThread - last statement it executes
        //--------------------------------------------------------------------//
        //this.stop();
        //--------------------------------------------------------------------//
    }
    
    
    
    /**
     * Starts a new thread an reads continuously from the sink firing
     * events when packets are received/unmarshaled.
     */
    public void start( )
    {
        Runnable runnable = new Runnable()
        {
            public void run()
            {
                while( stopped == false )
                {
                    try
                    {
                        List tempDispatchQueue = new ArrayList();
                        lock.lock();
                        try
                        {
                            // Make sure we have not received more
                            if( dispatchQueue.size() == 0 )
                            {
                                // Blocks until a message has been added to queue
                                messageReady.await();
                            }
                            
                            // Make copy so that queue can fill while callbacks
                            tempDispatchQueue.addAll(dispatchQueue);
                            dispatchQueue.clear();
                        }
                        finally
                        {
                            lock.unlock();
                        }
                        
                        // Empty the queue executing callbacks
                        //System.out.println( "    Queue size "+tempDispatchQueue.size() );
                        for( int i = 0; i < tempDispatchQueue.size(); i++ )
                        {
                            // FIFO order?
                            DispatchEntry entry = (DispatchEntry)tempDispatchQueue.get(i);
                            Header header       = entry.getHeader();
                            ByteBuffer inBuffer = entry.getInBuffer();
                            Sink sink           = entry.getSink();
                            
                            
                            //System.out.println("Dispatcher: received topic="+header.getTopic());
                            /*
                             * Filter - if a registered Message then unmarshal
                             * Otherwise simply forward
                             */
                            int topic = header.getTopic();
                            MessageBinding messageBinding = registry.getMessageBinding(topic);
                            if( messageBinding != null )
                            {
                                // This will takes a while - note header
                                // get unmarshaled twice
                                Message message=unmarshaler.unmarshal(header,
                                                                      inBuffer,
                                                                      messageBinding);
                                
                                // Notify listeners
                                fireRegisteredMessage( header, message, inBuffer, sink );
                            }
                            else
                            {
                                fireForwardMessage( header, inBuffer, sink );
                            }
                            
                            /*
                             * In one (easy to reason) place - return inBuffer
                             */
                            inPool.checkin(inBuffer);
                        }
                    }
                    catch( InterruptedException e )
                    {
                        // Catch condition variable exception - not normal
                        //fireExceptionReceived(e);
                        e.printStackTrace();
                    }
                    catch( Exception e )
                    {
                        // Catch user callback exceptions
                        //fireExceptionReceived(e);
                        e.printStackTrace();
                    }
                }
                System.out.println("=========================================");
                System.out.println("= Router Dispatcher is exitting - bye!");
                System.out.println("=========================================");
                stop(); // this simply sets the boolean state to stoppped
            }
        };
        Thread thread = new Thread(runnable);
        //thread.setDaemon(true);
        thread.setName( "DispatcherThread" );
        thread.start();
    }
    
    
    /**
     * Stops the ClientDispatcher from sending/receiving. Also results in
     * the sink being stopped.
     */
    public void stop( )
    {
        this.stopped = true;
        lock.lock();
        try
        {
            this.messageReady.signal();
        }
        finally
        {
            lock.unlock();
        }
    }
    
    /**
     * Determines if the dispatcher is still running.
     * @return true if stopped, false otherwise
     */
    public boolean isStopped()
    {
        return this.stopped;
    }
    
    /**
     * Solely to be used for debugging.
     * @param inBuffer
     * @param length
     */
    private static void print( ByteBuffer inBuffer, int length )
    {
        byte[] in = inBuffer.array();
        System.out.print( "position="+inBuffer.position() + " limit="+inBuffer.limit() );
        for (int j = 0; j < length; j++)
        {
            String s = Integer.toHexString( in[j]&0xff );
            if( s.length() == 1 ) s = "0"+s;
            System.out.print( " " + s );
        }
        System.out.println( "" );
    }
    
    
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param header
     * @param inBuffer
     * @param sink
     */
    public void fireForwardMessage( Header header,
                                    ByteBuffer inBuffer,
                                    Sink sink )
    {
        //Do this in order to only create Event once
        ForwardMessageEvent event = null;

        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.dispatchListeners.size(); i++)
        {
            // lazy initialization
            if(event == null)
            {
                event = new ForwardMessageEvent(  this,
                                         (long)System.currentTimeMillis(),
                                         header,
                                         inBuffer,
                                         sink
                                         );
            }
            // Get listener
            RouterDispatcherListener listener = ((RouterDispatcherListener)this.dispatchListeners.get(i));
            // Call listener
            listener.forwardMessageEvent( event );
        }
    }
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param header
     * @param inBuffer
     * @param sink
     */
    public void fireRegisteredMessage(  Header header,
                                        Message message,
                                        ByteBuffer inBuffer,
                                        Sink sink )
    {
        //Do this in order to only create Event once
        RouterMessageEvent event = null;

        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.dispatchListeners.size(); i++)
        {
            // lazy initialization
            if(event == null)
            {
                event = new RouterMessageEvent(  this,
                                         (long)System.currentTimeMillis(),
                                         header,
                                         message,
                                         inBuffer,
                                         sink
                                         );
            }
            // Get listener
            RouterDispatcherListener listener = ((RouterDispatcherListener)this.dispatchListeners.get(i));
            // Call listener
            listener.routerMessageEvent( event );
        }
    }
    
    /**
     * Adds to list of listeners. First checks to see if
     * listener is already in list, if it is it does not add it a second time.
     * @param listener to add.
     */
    public void addRouterDispatchListener( RouterDispatcherListener listener )
    {
        if( !this.dispatchListeners.contains(listener) )
        {
            this.dispatchListeners.add(listener);
        }
    }

    /**
     * Removes from list of listeners.
     * @param listener to remove.
     */
    public void removeRouterDispatchListener( RouterDispatcherListener listener )
    {
        this.dispatchListeners.remove(listener);
    }
    


}
